package network

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/terender/sfw/rlog"
	"gitee.com/terender/sfw/utils"
)

type mChan struct {
	sync.RWMutex
	closed  bool          // 是否已关闭
	channel chan *Message // 消息发送通道
}

func (c *mChan) Write(m *Message) error {
	c.RLock()
	defer c.RUnlock()
	if c.closed {
		return fmt.Errorf(`消息通道已经被关闭`)
	}
	c.channel <- m
	return nil
}

func (c *mChan) Close() {
	c.Lock()
	defer c.Unlock()
	if c.closed {
		return
	}
	c.closed = true
	close(c.channel)
}

// Connection 封装后的连接对象，内部封装了 net.Conn 对象
type Connection struct {
	server             *Server      // 服务器对象
	address            string       // 地址
	conn               *net.TCPConn // 连接实体
	gzip               int32        // 1为可以进行gzip压缩
	sessionID          int64        // 会话ID
	writeChan          *mChan       // 消息发送通道
	isClosing          int32        // 是否在主动关闭过程中
	isClosed           int32        // 是否完成关闭
	writeLoopClosed    int32        // 写线程是否退出
	readLoopClosed     int32        // 读线程是否退出
	pendingQuit        chan bool    // 完全退出通知
	OnConnectionClosed func(int64)  // 参数是sessionID
}

// newConnection 创建一个新的连接对象，创建接收缓冲，开启读线程和写线程
// 此方法只能由 Server 成功 Accept 一个新连接之后调用
func newConnection(server *Server, conn *net.TCPConn) *Connection {
	connection := &Connection{
		server:          server,
		address:         conn.RemoteAddr().String(),
		conn:            conn,
		gzip:            0,
		sessionID:       0,
		writeChan:       &mChan{channel: make(chan *Message, 32)},
		isClosing:       0,
		writeLoopClosed: 0,
		readLoopClosed:  0,
		pendingQuit:     make(chan bool),
	}

	go connection.writeLoop(connection.conn)
	go connection.readLoop(connection.conn)
	return connection
}

// Addr 远端地址
func (c *Connection) Addr() string { return c.address }

// GetSessionID 获取SessionID
func (c *Connection) GetSessionID() int64 { return atomic.LoadInt64(&c.sessionID) }

// SetSessionID 设置SessionID
func (c *Connection) SetSessionID(sessionid int64) { atomic.SwapInt64(&c.sessionID, sessionid) }

// SetGzip 设置Gzip
func (c *Connection) SetGzip(status int32) { atomic.SwapInt32(&c.gzip, status) }

// IsAlive 判断connection是否有效
func (c *Connection) IsAlive() bool {
	return atomic.LoadInt32(&c.isClosing) == 0 && c.IsReading() && c.IsWriting()
}
func (c *Connection) IsReading() bool { return atomic.LoadInt32(&c.readLoopClosed) == 0 }
func (c *Connection) IsWriting() bool { return atomic.LoadInt32(&c.writeLoopClosed) == 0 }

// 发送消息
func (c *Connection) Write(msg *Message) error {
	if msg == nil {
		return fmt.Errorf(`%s: Connection.Write error: msg cannot be null`, utils.FuncName())
	}
	defer func() {
		if err := recover(); err != nil {
			rlog.Errorf("%s", err)
		}
	}()
	return c.writeChan.Write(msg)
}

// Close 主动关闭连接
// 返回 pendingQuit channel，当 Connection 完成关闭操作后此 pendingQuit 也会被关闭
// 主动关闭从先关闭发送消息缓冲队列 writeChan 开始
// 写线程在将 writeChan 中已有的消息全部发送之后，关闭底层连接
// 然后读线程会产生读错误退出，退出时进行清理工作，并且关闭 pendingQuit channel
func (c *Connection) Close() <-chan bool {
	rlog.Debugf("主动关闭连接 [%v] [%v] [%+v]", c.address, c.sessionID, c.conn)
	atomic.StoreInt32(&c.isClosing, 1)
	c.conn.CloseRead()
	return c.pendingQuit
}

// [Deprecated]主动关闭连接
func (c *Connection) CloseConnection() <-chan bool { return c.Close() }

// 内部线程退出流程，读写线程退出时均需调用此方法，若读写线程均已退出，则完全关闭conn并且移除，关闭pendingQuit
func (c *Connection) onLoopQuit() {
	c.writeChan.Close()
	if atomic.LoadInt32(&c.readLoopClosed) != 1 ||
		atomic.LoadInt32(&c.writeLoopClosed) != 1 {
		return
	}
	if !atomic.CompareAndSwapInt32(&c.isClosed, 0, 1) {
		return
	}
	c.server.removeConnection(c.address)
	c.conn.Close()
	if c.OnConnectionClosed != nil {
		cb := c.OnConnectionClosed
		c.OnConnectionClosed = nil
		cb(c.sessionID)
	}
	close(c.pendingQuit)
}

// writeLoop 写线程处理函数
// 如果对连接的写操作遇到错误异常退出，或者 writeChan 被关闭，则主动关闭底层的连接，不进行清理工作
// 在读线程退出时进行清理工作
func (c *Connection) writeLoop(conn *net.TCPConn) {
	// 写线程退出时的清理工作
	defer func() {
		atomic.StoreInt32(&c.writeLoopClosed, 1)
		c.onLoopQuit()
	}()
	for msg := range c.writeChan.channel {
		buff := Marshal(msg, c.gzip)
		for {
			n, err := conn.Write(buff)
			if err != nil {
				ne, ok := err.(net.Error)
				if ok && ne.Temporary() {
					continue
				}
				return
			}
			if n >= len(buff) {
				break
			}
			buff = buff[n:]
		}
	}
}

// readLoop 读线程处理函数
func (c *Connection) readLoop(conn *net.TCPConn) {
	// 读线程退出时的清理工作
	defer func() {
		atomic.StoreInt32(&c.readLoopClosed, 1)
		c.onLoopQuit()
	}()

	dataLen := 0
	defaultBuff := make([]byte, c.server.ReadBuffSize)
	var extraBuff []byte

	/// 最开始使用默认缓冲区接受数据
	inuseBuff := defaultBuff
	var initial = true
	for {
		readlen := 0
		var err error
		if initial {
			conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(c.server.ReadTimeoutInitial)))
			initial = false
		} else {
			conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(c.server.ReadTimeout)))
		}
		readlen, err = conn.Read(inuseBuff[dataLen:])
		if err != nil {
			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				if ne.Timeout() {
					rlog.Errorf("read 超时: %s", ne.Error())
					return
				}
				continue
			}
			// 远端正常关闭连接
			if err == io.EOF {
				return
			}
			rlog.Errorf("其他read错误:%s", err.Error())
			return
		}

		dataLen += readlen

		//log.Println("//////此次read共读取[", readlen, "]字节消息，读缓冲区头指针位置[", hp, "],尾指针位置[", tp, "],工作指针位置[", curp, "]//////")
		left, flex := c.handle(inuseBuff[:dataLen])

		/// 未发生消息截断，缓冲区中的数据全部处理完毕
		if left == 0 {
			dataLen = 0
			extraBuff = nil
			inuseBuff = defaultBuff
			continue
		}

		/// 发生了消息截断
		/// 判断是否需要更换缓冲区，有三种情况需要更换
		/// 1. 上次使用默认缓冲，下次用额外缓冲
		/// 2. 上次使用额外缓冲，下次需要的缓冲长度超过上次，重新开辟缓冲
		/// 3. 上次使用额外缓冲，下次使用默认缓冲
		nextBuff := defaultBuff
		buffChanged := false
		if flex > len(defaultBuff) {
			if flex > c.server.ReadBuffSizeMax {
				rlog.Errorf("Read 错误: 消息长度超过接收缓冲区最大长度")
				return
			}
			if extraBuff == nil || len(extraBuff) < flex {
				extraBuff = make([]byte, getNearestPOT(flex))
				buffChanged = true
			}
			nextBuff = extraBuff
		} else if extraBuff != nil {
			buffChanged = true
			extraBuff = nil
		}

		/// 如果截断的消息不在缓冲区头部，或者更换了缓冲区，都需要将阶段消息拷贝至目标缓冲区头部
		if left < dataLen || buffChanged {
			copy(nextBuff, inuseBuff[dataLen-left:dataLen])
		}
		inuseBuff = nextBuff
		dataLen = left
		////log.Println("//////复位指针，读缓冲区头指针位置[", hp, "],尾指针位置[", tp, "],工作指针位置[", curp, "]//////")
	}
}

// handle 处理缓冲区里的数据
// 按照消息头制定长度截取消息
// 返回剩余无法处理的数据长度以及被截断的消息的应收长度
func (c *Connection) handle(buff []byte) (left int, flex int) {
	rptr := 0
	for rptr < len(buff) {
		var msg *Message
		var err error
		msg, flex, err = Unmarshal(buff[rptr:])
		if err != nil {
			//rlog.Errorf("connection[%d] err: %s", atomic.LoadInt64(&c.sessionID), err.Error())
			break
		}
		rptr += int(msg.MsgHead.Len)

		if msg.Content != nil && len(msg.Content) > 0 && msg.MsgHead.Version == 1 {
			var b = bytes.NewBuffer(msg.Content)
			r, err := gzip.NewReader(b)
			if err != nil {
				rlog.Errorf(err.Error())
				c.server.defaultRoutine.processMessage(&Context{
					conn:    c,
					msg:     msg,
					Time:    time.Now(),
					cache:   make(map[string]interface{}),
					routine: c.server.defaultRoutine,
				})
				continue
			}
			undatas, _ := ioutil.ReadAll(r)
			r.Close()
			msg.Content = undatas
		}
		routine, ok := c.server.findMessageRoute(msg.MsgHead.ID)
		if !ok {
			rlog.Errorf(`Unkown message id [%v]`, msg.MsgHead.ID)
			routine = c.server.defaultRoutine
		}
		routine.processMessage(&Context{
			conn:    c,
			msg:     msg,
			Time:    time.Now(),
			cache:   make(map[string]interface{}),
			routine: routine,
		})
	}
	//log.Println(utils.FuncName(), `: goID[`, base.Goid(), "] //////此次read结束，读缓冲区头指针位置[", hp, "],尾指针位置[", tp, "],工作指针位置[", curp, "]//////")

	return len(buff) - rptr, flex
}

func getNearestPOT(loss int) int {
	temp := loss - 1
	temp |= temp >> 1
	temp |= temp >> 2
	temp |= temp >> 4
	temp |= temp >> 8
	temp |= temp >> 16
	cap := 0
	if temp < 0 {
		cap = 1
	} else {
		cap = temp + 1
	}
	return cap
}
